package schedule

import (
	"fmt"
	"net"
	"sync"
	"time"
)

import (
	"code.google.com/p/gcfg"
)

import (
	"r_log"
	"recom"
	"tp"
	"util"
)

type ScheduleSocketConf struct {
	Port              int
	DeadInterval      time.Duration
	RetryInterval     time.Duration
	KeepAliveInterval time.Duration
	RetryTimes        int
}

type ScheduleServer struct {
	Conn     net.Conn
	ConnLock sync.Mutex
	SSConf   ScheduleSocketConf

	Connected bool
	Recomer   *recom.Recommender
	State     int32
	AckBody   []byte
	StateLock sync.Mutex
}

//var SS *ScheduleServer

func InitScheduleServer(confPath string, r *recom.Recommender) (*ScheduleServer, error) {
	ss := new(ScheduleServer)
	err := gcfg.ReadFileInto(ss, confPath)
	if err != nil {
		return nil, fmt.Errorf("gcfg.ReadFileInto(): %s", err.Error())
	}
	ss.Connected = false
	ss.Recomer = r
	ss.State = QUANTTRADING_CALCRESULT__UNKNOWN
	return ss, nil
}

func (ss *ScheduleServer) Run() {
	go ss.keepalive()
	ss.StartServer()
}

func (ss *ScheduleServer) StartServer() {
	l, _ := net.Listen("tcp", fmt.Sprintf(":%v", ss.SSConf.Port))
	for {
		c, err := l.Accept()
		if err != nil {
			r_log.Logger.Error("ss.Startserver():", err)
			continue
		}
		//处理
		ss.Conn = c
		ss.Connected = true
		ss.receivePacket(c)
	}
}

func (ss *ScheduleServer) receivePacket(c net.Conn) {
	defer c.Close()
	key := c.RemoteAddr().String()
	bhead := make([]byte, QTPacketHeadSize)
	for {
		c.SetReadDeadline(time.Now().Add(time.Second * ss.SSConf.DeadInterval))
		nhead := 0
		for nhead < QTPacketHeadSize {
			m, err := c.Read(bhead[nhead:])
			if err != nil {
				r_log.Logger.Error("receivePacketHead(): time=%v,key=%v, err:%v", time.Now().String(), key, err)
				ss.Connected = false
				return
			}
			nhead += m
		}
		var head SQuantTradingPacketHead
		head.DeCode(bhead)
		var bbody []byte
		if head.DwBodyLen > 0 {
			bbody = make([]byte, head.DwBodyLen)
			nbody := 0
			for nbody < int(head.DwBodyLen) {
				m, err := c.Read(bbody[nbody:])
				if err != nil {
					r_log.Logger.Error("receivePacketBody(): time=%v,key=%v, err:%v", time.Now().String(), key, err)
					ss.Connected = false
					return
				}
				nbody += m
			}
		}
		switch head.WPid {
		case QUANTTRADING_PID__KEEP_ALIVE:
			//fmt.Println("QUANTTRADING_PID__KEEP_ALIVE", head)
		case QUANTTRADING_PID__STKRECOMM_CALC:
			var body SQuantTradingBody_StkRecommCalc
			body.DeCode(bbody)
			r_log.Logger.Info("QUANTTRADING_PID__STKRECOMM_CALC: ", body)
			ss.StateLock.Lock()
			if ss.State == QUANTTRADING_CALCRESULT__UNKNOWN {
				go ss.dealPacket_StkRecommCalc(body)
			} else {
				if ss.State != QUANTTRADING_CALCRESULT__WAIT {
					r_log.Logger.Info("QUANTTRADING_PID__STKRECOMM_CALC: unexpect order! state is %v now!", ss.State)
				}
			}
			ss.StateLock.Unlock()
		case QUANTTRADING_PID__STKRECOMM_INIT_CALCDATA:
			var body SQuantTradingBody_StkRecommInitCalcData
			body.DeCode(bbody)
			r_log.Logger.Info("QUANTTRADING_PID__STKRECOMM_INIT_CALCDATA: ", body)
			ss.StateLock.Lock()
			if ss.State == QUANTTRADING_CALCRESULT__UNKNOWN {
				go ss.dealPacket_StkRecommInitCalcData(body)
			} else {
				if ss.State != QUANTTRADING_INITRDATA__WAIT {
					r_log.Logger.Info("QUANTTRADING_PID__STKRECOMM_INIT_CALCDATA: unexpect order! state is %v now!", ss.State)
				}
			}
			ss.StateLock.Unlock()

		}
	}
}

func (ss *ScheduleServer) sendPacket(packet []byte) error {
	length := len(packet)
	n := 0
	ss.ConnLock.Lock()
	defer ss.ConnLock.Unlock()
	for n < length {
		m, err := ss.Conn.Write(packet[n:])
		if err != nil {
			return fmt.Errorf("c.Write(): %v", err)
		}
		n += m
	}
	return nil
}

func (ss *ScheduleServer) keepalive() {
	for {
		time.Sleep(time.Second * ss.SSConf.KeepAliveInterval)
		if ss.Connected {
			var body []byte
			pid := QUANTTRADING_PID__KEEP_ALIVE
			ss.StateLock.Lock()
			switch ss.State {
			case QUANTTRADING_INITRDATA__WAIT:
				pid = QUANTTRADING_PID__STKRECOMM_INIT_CALCDATA
				body = ss.AckBody
			case QUANTTRADING_CALCRESULT__WAIT:
				pid = QUANTTRADING_PID__STKRECOMM_CALC
				body = ss.AckBody
			default:
				body = []byte{}
			}
			ss.StateLock.Unlock()

			err := ss.SendPacket(body, pid)
			if err != nil {
				r_log.Logger.Error("keepalive(wpid=%v): %v", pid, err)
				continue
			}
		}
	}
}

func (ss *ScheduleServer) SendPacket(body []byte, pid uint16) error {
	var head SQuantTradingPacketHead
	head.WPid = pid
	head.DwBodyLen = uint32(len(body))
	bhead := head.EnCode()
	b := append(bhead, body...)
	err := ss.sendPacket(b)
	if err != nil {
		return fmt.Errorf("ss.SendPacket(): %v", err)
	}
	return nil
}

func (ss *ScheduleServer) dealPacket_StkRecommCalc(packet SQuantTradingBody_StkRecommCalc) {
	var body SQuantTradingBody_StkRecommCalcA
	body.UllQuantCalcExecId = packet.UllQuantCalcExecId
	body.UllQuantCalcPoint = packet.UllQuantCalcPoint
	/////
	if ss.Recomer.LastCalcPoint < util.TimeStamp(packet.UllQuantCalcPoint) {
		batch := [util.PERIOD_COUNT]int32{-1, -1, -1, -1}
		for i := 0; i < int(packet.YCalcKTypeItemCount); i++ {
			batch[util.Period(packet.ArrCalcKTypeItem[i].YAccountKType)] = int32(packet.ArrCalcKTypeItem[i].DwCalcBatch)
		}
		ss.StateLock.Lock()
		ss.State = QUANTTRADING_CALCRESULT__WAIT
		body.NCalcResultCode = QUANTTRADING_CALCRESULT__WAIT
		ss.AckBody = body.EnCode()
		ss.StateLock.Unlock()
		ss.Recomer.Tps = tp.NewTradingPosition(util.TimeStamp(packet.UllQuantCalcPoint), util.DateAndTime(packet.ArrCalcKTypeItem[0].DwCalcTradeDate), batch, ss.Recomer.Res)
		ss.Recomer.ResetStatistic(util.TimeStamp(packet.UllQuantCalcPoint))
		err := ss.Recomer.Tps.Init()
		if err != nil {
			r_log.Logger.Error(err)
			body.NCalcResultCode = QUANTTRADING_CALCRESULT__FAIL
		} else {
			ss.Recomer.Recommender()
			ss.Recomer.SaveResults()
			ss.Recomer.Tps.SaveAccState()
			ss.Recomer.UpdateLastCalcPiont()
			body.NCalcResultCode = QUANTTRADING_CALCRESULT__SUCC
		}
		ss.StateLock.Lock()
		ss.State = QUANTTRADING_CALCRESULT__UNKNOWN
		ss.AckBody = []byte{}
		ss.StateLock.Unlock()
	} else {
		body.NCalcResultCode = QUANTTRADING_CALCRESULT__SUCC
	}
	ss.SendPacket(body.EnCode(), QUANTTRADING_PID__STKRECOMM_CALC)
}

func (ss *ScheduleServer) dealPacket_StkRecommInitCalcData(packet SQuantTradingBody_StkRecommInitCalcData) {
	var body SQuantTradingBody_StkRecommInitCalcDataA
	body.DwCalcTradeDate = packet.DwCalcTradeDate
	///
	ss.StateLock.Lock()
	ss.State = QUANTTRADING_INITRDATA__WAIT
	body.NCalcResultCode = QUANTTRADING_INITRDATA__WAIT
	ss.AckBody = body.EnCode()
	ss.StateLock.Unlock()
	ss.Recomer.ResetResult()
	ss.Recomer.Res.Update()

	ss.StateLock.Lock()
	ss.State = QUANTTRADING_CALCRESULT__UNKNOWN
	ss.AckBody = []byte{}
	ss.StateLock.Unlock()
	body.NCalcResultCode = QUANTTRADING_CALCRESULT__SUCC
	ss.SendPacket(body.EnCode(), QUANTTRADING_PID__STKRECOMM_INIT_CALCDATA)
}
